package cn.tedu.charging.order.consumer;

import cn.tedu.charging.common.consts.ChargingTopicConstants;
import cn.tedu.charging.common.entity.ChargingProgressEntity;
import cn.tedu.charging.common.enums.GunStatusEnum;
import cn.tedu.charging.common.message.ChargingProgressMsg;
import cn.tedu.charging.common.message.CheckResultMsg;
import cn.tedu.charging.common.param.ChargingProgressParam;
import cn.tedu.charging.common.param.GunStatusUpdateParam;
import cn.tedu.charging.common.vo.ChargingProgressVO;
import cn.tedu.charging.order.context.MqttContext;
import cn.tedu.charging.order.feign.CostClient;
import cn.tedu.charging.order.feign.DeviceClient;
import cn.tedu.charging.order.po.BillFailPO;
import cn.tedu.charging.order.po.BillSuccessPO;
import cn.tedu.charging.order.repository.BillFailRepository;
import cn.tedu.charging.order.repository.BillSuccessRepository;
import cn.tedu.charging.order.server.WebSocketServer;
import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.BeanUtils;

import java.math.BigDecimal;
import java.util.Date;
import java.util.Random;


@Slf4j
public class OrderEmqxConsumer implements MqttCallbackExtended {
    private MqttClient mqttClient;
    //mqtt消费者上下文对象 主要就是消费者传递容器bean对象
    private MqttContext mqttContext;

    public OrderEmqxConsumer(MqttClient mqttClient,MqttContext mqttContext){
        this.mqttClient=mqttClient;
        this.mqttContext=mqttContext;
    }
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        try {
            //一旦成功连接 使用客户端同时监听2个主题
            mqttClient.subscribe(ChargingTopicConstants.CHARGING_PROGRESS_TOPIC);
            mqttClient.subscribe(ChargingTopicConstants.PILE_CALLBACK_TOPIC);
        } catch (MqttException e) {
            log.info("监听订阅主题失败",e);
        }
    }

    @Override
    public void connectionLost(Throwable cause) {
        log.error("监听订阅主题链接断开",cause);
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.info("消费者接受消息,主题:{},消息:{}",topic,message.toString());
        //解析消息 交给容器业务对象处理数据
        try{
            //由于消息可能来自于不同的主题,if判断
            if (topic!=null&&topic.equals(ChargingTopicConstants.PILE_CALLBACK_TOPIC)){
                //充电桩自检结果
                handleCheckResult(topic,message);
            }else if(topic!=null&&topic.equals(ChargingTopicConstants.CHARGING_PROGRESS_TOPIC)){
                handleChargingProgress(topic,message);
            }
        }catch (Exception e){
            log.error("消费出现异常",e);
        }

    }
    //消费者中 重点关注的 处理充电进度的业务逻辑
    private void handleChargingProgress(String topic, MqttMessage message) {
        log.info("处理充电进度:{}",message);
        //1.解析消息对象
        String json = message.toString();
        ChargingProgressMsg chargingProgressMsg = JSON.parseObject(json, ChargingProgressMsg.class);
        //2.调用计价中心接口 获取价格
        ChargingProgressVO chargingVO=calculate(chargingProgressMsg);
        //3.将同步数据 和计价结果存储到es中
        ChargingProgressEntity entity=saveESChargingEntity(chargingProgressMsg,chargingVO);
        //4.将最终的结果同步给客户端小程序 webSocketServer 知道userId是谁 约定好和前端交互的接口
        WebSocketServer webSocketServer = mqttContext.getWebSocketServer();
        //就可以将同步的消息进度发送会给前端展示了
        //给消息的来源中携带的userId 发送消息
        String jsonString = JSON.toJSONString(entity);
        log.info("即将给用户:{},发送消息:{}",chargingProgressMsg.getUserId(),jsonString);
        webSocketServer.pushMessage(jsonString,chargingProgressMsg.getUserId());
    }

    private ChargingProgressEntity saveESChargingEntity(ChargingProgressMsg chargingProgressMsg, ChargingProgressVO chargingVO) {
        //需要考虑这个id值 是一个时序数据
        //充电的进度 是有时间先后顺序 展示查询页必须有一个时序依据,不能展示的时候将数据按照
        //18:03:25 第一次进度
        //18:03:45 第二次进度
        //18:04:05 第三次进度
        //上述循序不能颠倒,可以利用存储在es的id值
        //可以使用当前系统的毫秒数生成id值,为了避免同一时刻id相同添加一个随机数4位
        Long id=Long.valueOf(System.currentTimeMillis()+""+(new Random().nextInt(9000)+1000));
        ChargingProgressEntity entity = new ChargingProgressEntity();
        BeanUtils.copyProperties(chargingProgressMsg,entity);
        entity.setId(id);
        entity.setTotalCost(chargingVO.getTotalCost());
        mqttContext.getChargingProgressRepository().save(entity);
        return entity;
    }

    private ChargingProgressVO calculate(ChargingProgressMsg chargingProgressMsg) {
        //返回一个假的数据
        //1.准备一个计算价格的入参 来自于msg
        ChargingProgressParam param = new ChargingProgressParam();
        param.setBillId(chargingProgressMsg.getBillId());
        param.setUserId(chargingProgressMsg.getUserId());
        param.setStationId(chargingProgressMsg.getStationId());
        param.setGunId(chargingProgressMsg.getGunId());
        param.setChargingCapacity(chargingProgressMsg.getChargingCapacity());
        //2.从mqttContext拿到客户端
        CostClient costClient = mqttContext.getCostClient();
        ChargingProgressVO voData = costClient.calculate(param).getData();
        return voData;
    }

    private void handleCheckResult(String topic, MqttMessage message) {
        BillFailRepository billFailRepository = mqttContext.getBillFailRepository();
        BillSuccessRepository billSuccessRepository = mqttContext.getBillSuccessRepository();
        DeviceClient deviceClient = mqttContext.getDeviceClient();
        //1.解析消息成为对象
        String json = message.toString();
        log.info("订单接收设备反馈的结果是:{}",json);
        CheckResultMsg checkResultMsg = JSON.parseObject(json, CheckResultMsg.class);
        //先查再写
        String billId = checkResultMsg.getOrderNo();
        boolean exists=false;
        //执行查询方法 success和fail
        exists=checkFailExists(billId)||checkSuccessExists(billId);
        if (exists){
            //说明当前环境已经具备了订单记录不能再新增
            return;
        }
        //读取success
        boolean success = checkResultMsg.isSuccess();
        log.info("订单接收设备反馈的结果是:{}",success);
        //2判断 如果true 新增success 如果是false 新增fail
        if (success){
            BillSuccessPO poParam=assembleSuccessPO(checkResultMsg);
            billSuccessRepository.save(poParam);
        }else{
            BillFailPO poParam=assembleFailPO(checkResultMsg);
            billFailRepository.save(poParam);
            //远程调用修改枪状态为 检查失败 故障 openFeign
            GunStatusUpdateParam param=new GunStatusUpdateParam();
            param.setStatus(GunStatusEnum.FAULT.getStatus());
            param.setGunId(checkResultMsg.getGunId());
            deviceClient.updateGunStatus(param);
            //调用WEBSOCKET 给用户通知换枪 并且结合优惠券业务发送一个打折优惠券
            WebSocketServer webSocketServer = mqttContext.getWebSocketServer();
            //TODO 和前端联调 再修正
            webSocketServer.pushMessage("对不起 充电桩故障请换枪",checkResultMsg.getUserId());
        }

    }

    private boolean checkSuccessExists(String billId) {
        //select count from table wherer bill_id=#{}
        QueryWrapper queryWrapper=new QueryWrapper();
        queryWrapper.eq("bill_id",billId);
        long count = mqttContext.getBillSuccessRepository().count(queryWrapper);
        return count>0;
    }

    private boolean checkFailExists(String billId) {
        QueryWrapper queryWrapper=new QueryWrapper();
        queryWrapper.eq("bill_id",billId);
        long count = mqttContext.getBillFailRepository().count(queryWrapper);
        return count>0;
    }

    private BillFailPO assembleFailPO(CheckResultMsg checkResultMsg) {
        BillFailPO po=new BillFailPO();
        BeanUtils.copyProperties(checkResultMsg,po);
        //补充其他属性
        po.setBillId(checkResultMsg.getOrderNo());
        po.setCreateTime(new Date());
        po.setUpdateTime(new Date());
        return po;
    }

    private BillSuccessPO assembleSuccessPO(CheckResultMsg checkResultMsg) {
        BillSuccessPO po=new BillSuccessPO();
        //直接拷贝 gunId pileId stationId operationId userId
        BeanUtils.copyProperties(checkResultMsg,po);
        //billId orderNo
        po.setBillId(checkResultMsg.getOrderNo());
        //时间相关
        po.setCreateTime(new Date());
        po.setUpdateTime(new Date());
        return po;
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.info("消费者消费消息 投递结束");
    }
}
